from bokeh.plotting import figure
from bokeh.io import output_notebook, show
import pandas
import jinja2
from bokeh.palettes import Spectral6, Spectral10
from bokeh.io import curdoc
from bokeh.models import ColumnDataSource, DatetimeTickFormatter, Select
from bokeh.models import DataTable, DateFormatter, TableColumn, HTMLTemplateFormatter
from bokeh.models import CrosshairTool, HoverTool, CustomJSHover, Span, Range1d, FixedTicker
from bokeh.models import Div, CheckboxButtonGroup, CustomJS, Spacer
from bokeh.layouts import layout
from bokeh.plotting import figure, reset_output
from datetime import datetime
from math import radians # rotate axis ticks
import numpy as np
import pandas as pd
import os
import json
from scipy.signal import butter,filtfilt
reset_output()
output_notebook()
plot formatting/structure
DEFAULT_PLOT_HEIGHT = 500
DEFAULT_PLOT_WIDTH = 1000
# Define the number of plots and which data lines are included within each plot
PLOT_COMPOSITION = [
{"plot_name": "State",
"metrics_included": ["state"],
"height":170,
"state_plot": True,
"yaxis_label":"State",
},
{"plot_name": "Temperature",
"metrics_included": ["tw1", "tw2", "tw3", "tc1", "tc2", "tc3","tg", "ti", "to", "tv"],
"yaxis_label":"Temperature [ C ]",
},
{"plot_name": "Absolute Pressure",
"metrics_included": ["pg", "po", "pv","pi"],
"yaxis_label":"Pressure [ kPag ]"
},
{"plot_name": "Differential Pressure",
"metrics_included": ["dpo", "dpf", "dpv"],
"yaxis_label":"Pressure [ kPa ]"
},
{"plot_name": "Mass Flow",
"metrics_included": ["mdoti", "mdoto"],
"yaxis_label":"Mass Flow [ kg/s ]"
},
{"plot_name": "CO2 Concentration",
"metrics_included": ["gai", "gao"],
"yaxis_label":"Volumetric Concentration [ %C02 ]"
}
]
function definitions
def str_to_datetime(s):
"""Convert a string to a datetime object"""
return datetime.strptime(s, "%Y-%m-%d-%H-%M-%S")
def get_num_rows(csv_path):
"""Return the number of rows in a csv file given the file path"""
with open(csv_path, 'r') as infile:
raw_lines = infile.readlines()
return len(raw_lines)
def format_df_to_source(df):
"""Convert a Pandas dataframe to the format that Bokeh accepts as a source"""
return {col_name: list(col_data) for (col_name, col_data) in df.items()}
read to dataframe
with open("data_format.json", "r") as infile:
metric_names = [item["metric_name"] for item in json.load(infile)]
data_log_path = "data_analysis\\FirstCarbonation_2022-11-25-14-26-40.csv"
existing_df = pandas.read_csv(data_log_path)
existing_df["datetime"] = existing_df["datetime"].map(str_to_datetime)
source = ColumnDataSource(format_df_to_source(existing_df))
plotting related tools
# plot_layout = [[data_table]]
cursor_x = Span(dimension="width", line_dash="dotted", line_width=2)
cursor_y = Span(dimension="height", line_dash="dashed", line_width=2)
TOOLS = "pan,wheel_zoom,box_zoom,save,reset"
# This formatter is used for the time that's shown on the hover tooltip
time_custom_formatter = CustomJSHover(code="""
const x = special_vars.x
// Create a new JavaScript Date object based on the timestamp
// the argument is in milliseconds
var date = new Date(x);
// Hours part from the timestamp
var hours = date.getHours();
// Minutes part from the timestamp
var minutes = "0" + date.getMinutes();
// Seconds part from the timestamp
var seconds = "0" + date.getSeconds();
// Will display time in 10:30:23 format
var formattedTime = hours + ':' + minutes.substr(-2) + ':' + seconds.substr(-2);
return formattedTime
""")
HOVER_TOOL = HoverTool(
tooltips=[
( 'value', '$snap_y' ),
( 'time', '@datetime{custom}' ),
],
formatters={
'@datetime' : time_custom_formatter,
},
# display a tooltip whenever the cursor is vertically in line with a glyph
mode='vline'# https://docs.bokeh.org/en/latest/docs/user_guide/interaction/tools.html#hit-testing-behavior
)
Define a filter for pressure data
# Filter requirements.
T = 0.5 # Sample Period
fs = 2 # sample rate, Hz
cutoff = 0.025 # desired cutoff frequency of the filter, Hz , slightly higher than actual 1.2 Hz
nyq = 0.5 * fs # Nyquist Frequency
order = 1 # sin wave can be approx represented as quadratic
def butter_lowpass_filter(data, cutoff, fs, order):
normal_cutoff = cutoff / nyq
# Get the filter coefficients
b, a = butter(order, normal_cutoff, btype='low', analog=False)
y = filtfilt(b, a, data)
return y
Main plotting loop
first_plot_flag = True
for each_plot in PLOT_COMPOSITION:
# Create figure
if first_plot_flag:
p = figure(x_axis_type="datetime",
width=DEFAULT_PLOT_WIDTH,
height=each_plot.get("height", DEFAULT_PLOT_HEIGHT),
tools=TOOLS
)
else:
# linking x range with the first plot: https://docs.bokeh.org/en/latest/docs/user_guide/interaction/linking.html#linked-panning
p = figure(x_axis_type="datetime",
width=DEFAULT_PLOT_WIDTH,
height=each_plot.get("height", DEFAULT_PLOT_HEIGHT),
tools=TOOLS,
x_range=first_x_range
)
# Add hover tooltip
p.add_tools(HOVER_TOOL)
# Add crosshair tool
p.add_tools(CrosshairTool(overlay=[cursor_x, cursor_y])) # https://docs.bokeh.org/en/latest/docs/user_guide/interaction/linking.html#linked-crosshair
# Autohide toolbar
p.toolbar.autohide = True
# Plot lines with markers
# Note: the N in "SpectralN" indicates the number of colors in the palette. If there will be more than 6 lines in a plot, N needs to be increased, otherwise the zip function will yield maximally only 6 pairs of items
lowpass_color_index = 6
for metric_name, color, in zip(each_plot["metrics_included"], Spectral10):
p.line(x="datetime",y=metric_name, source=source, color=color, legend_label=metric_name)
# p.circle(x="datetime",y=metric_name, source=source, color=color, legend_label=metric_name)
if each_plot["plot_name"] == "Absolute Pressure" or each_plot["plot_name"] == "Differential Pressure":
filtered_data = butter_lowpass_filter(existing_df[metric_name], cutoff, fs, order)
legend_lowpass = metric_name + "_lowpass"
p.line(x=existing_df["datetime"],y=filtered_data,color = Spectral10[lowpass_color_index],legend_label = legend_lowpass)
lowpass_color_index = lowpass_color_index + 1
date_pattern = "%Y-%m-%d\n%H:%M:%S"
p.xaxis.formatter = DatetimeTickFormatter(
seconds = date_pattern,
minsec = date_pattern,
minutes = date_pattern,
hourmin = date_pattern,
hours = date_pattern,
days = date_pattern,
months = date_pattern,
years = date_pattern
)
p.title.text = each_plot["plot_name"]
p.xaxis.major_label_orientation=radians(80)
# change to vary based on plot
p.yaxis.axis_label = each_plot["yaxis_label"]
# p.yaxis.axis_label = "Value" # Use newer syntax: https://stackoverflow.com/questions/24131501/x-and-y-axis-labels-for-bokeh-figure
p.legend.click_policy="hide" # https://docs.bokeh.org/en/latest/docs/user_guide/interaction/legends.html#hiding-glyphs
p.legend.location = "top_left"
# p.legend.click_policy="hide" # https://docs.bokeh.org/en/latest/docs/user_guide/interaction/legends.html#hiding-glyphs
if "state_plot" in each_plot:
p.y_range = Range1d(0, 4)
p.yaxis.ticker = FixedTicker(ticks=[0, 1, 2, 3, 4])
show(p)
# plot_layout.append([p])
if first_plot_flag:
first_x_range = p.x_range
first_plot_flag = False